package com.talent.study.projectreactor;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;

import java.time.Duration;
import java.util.function.Function;
import java.util.stream.IntStream;

/**
 * @ClassName：Main
 * @Author: hszhou
 * @Date: 2025/1/18 9:58
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class Main {

    public static void main(String[] args) throws InterruptedException {

        test04();
       // test04();

        Thread.sleep(10000);
    }

    private static void test04(){

        Flux.just("tom")
                .map(s -> {
                    System.out.println("[map] Thread name: " + Thread.currentThread().getName());
                    return s.concat("@mail.com");
                })
                .publishOn(Schedulers.newParallel("thread-publishOn"))
                .map(e->{
                    System.out.println("[map2] Thread name: " + Thread.currentThread().getName());
                    return e;
                })
                .filter(s -> {
                    System.out.println("[filter] Thread name: " + Thread.currentThread().getName());
                    return s.startsWith("t");
                })

                .subscribeOn(Schedulers.newParallel("thread-subscribeOn"))
                .subscribe(s -> {
                    System.out.println("[subscribe] Thread name: " + Thread.currentThread().getName());
                    System.out.println(s);
                });


    }

    private static void test01(){

        Flux.push(fluxSink -> {
            System.out.println(Thread.currentThread().getName());
            //可以使用单线程的方式 调用其他的方法，其他方法的返回值 可以通过fluxSink追加到下游响应式流中
            IntStream.range(100,300)
                    .forEach(fluxSink::next);
        }).delayElements(Duration.ofMillis(100))
                .subscribe(System.out::println);

        System.out.println("-----");
    }

    private static void test02(){


      Flux.just("1","2")
              .doOnNext(e-> System.out.println(e))
              .thenMany(Flux.just("3"))
              .doOnNext(s-> System.out.println(s))
              .then();
              //.subscribe(s-> System.out.println(s));
    }

    /**
     * 组合和转换响应式流
     */
    private static void test03(){

      //  Flux.defer(()->Flux.just("111"));
//        Flux<String> just = Flux.just("11");
//        ConnectableFlux<String> publish = just.publish();

        Hooks.onOperatorDebug();
        Function<Flux<String>,Flux<String>> logUserInfo = stream->
        stream.index().doOnNext(tp-> System.out.println("["+tp.getT1()+"]User"+tp.getT2())).map(Tuple2::getT2);

        Flux.range(1000,3)
                .map(i->"user-"+i)
                .transform(logUserInfo)
                .subscribe(e-> System.out.println(e));

//        try with sources
//
//        Flux.using(
//                //创建资源
//                //使用资源
//                //关闭资源
//        ).subscribe();
    }
}
